/*
 * @Author: wuqingchun
 * @Date: 2024-10-14 15:18:53
 * @Description: 
 * @LastEditors: wuqingchun
 * @LastEditTime: 2024-10-16 10:02:28
 */

#include <sys/prctl.h>
#include <sys/select.h>
#include <sys/socket.h>

#include "redis_client.h"


std::atomic<RedisClientState> CRedisClient::m_atmState {RedisClientState::REDIS_STATE_UNINT};

CRedisClient::CRedisClient()
{

}

CRedisClient::~CRedisClient()
{
    if (m_pstRedisCtx != nullptr)
    {
        redisFree(m_pstRedisCtx);
        m_pstRedisCtx = nullptr;
    }
}

int32_t CRedisClient::Start(const char* pcHost, uint16_t usPort)
{
    m_strHost = pcHost;
    m_usPort = usPort;
    
    // int32_t iErr = __Connect();
    // if (iErr != 0)
    // {
    //     return iErr;
    // }

    printf("Start redis client event thread ...\n");
    m_bConnMgrRunning = true;
    m_objConnMgrThread = std::thread(&CRedisClient::__ThreadConnectMgrFunc, this);
    m_atmState.store(RedisClientState::REDIS_STATE_READY);

    return 0;
}

void CRedisClient::Stop()
{
    m_bConnMgrRunning = false;

    if (m_objConnMgrThread.joinable())
    {
        m_objConnMgrThread.join();
    }

    printf("Stop redis client\n");
}

int32_t CRedisClient::PushMarketData(const tick& refTick)
{
    if (m_atmState.load() != RedisClientState::REDIS_STATE_CONNECTED)
    {
        return -1;
    }

    char szMarketData[4096];
    char szTimeStamp[32];
    char szKey[32];

    snprintf(szKey, 32, "%s_%06d", refTick.instrument_id, refTick.tradingday);
    snprintf(szTimeStamp, 16, "%d%03hd", refTick.time, refTick.milli);
    snprintf(szMarketData, 4096, "%.03lf,%ld,%ld,%.03lf", refTick.last_price, refTick.openinterest, refTick.volume, refTick.turnover);

    printf("zadd command 'zadd %s %s %s'\n", szKey, szTimeStamp, szMarketData);

    redisReply* pstReply;
    
    {
        std::unique_lock<std::mutex> objUL(m_objMtx);
        // zadd key score member
        pstReply = (redisReply*)redisCommand(m_pstRedisCtx, "zadd %s %s %s", szKey, szTimeStamp, szMarketData);
    }
    
    if (pstReply == NULL)
    {
        printf("zadd command failed: I/O error\n");
        return -1;
    }

    if (pstReply->type == REDIS_REPLY_ERROR) 
    {
        printf("zadd failed: %s\n",pstReply->str);
        return -1;
    }

    printf("zadd command successfully\n");

    return 0;
}

int32_t CRedisClient::GetLatestMarketData(tick& refTick)
{
    if (m_atmState.load() != RedisClientState::REDIS_STATE_CONNECTED)
    {
        return -1;
    }

    redisReply* pstReply;

    char acMarketData[4096];
    char szKey[32];


    snprintf(szKey, 32, "%s_%06d", refTick.instrument_id, refTick.tradingday);

    {
        std::unique_lock<std::mutex> objUL(m_objMtx);
        // zadd key score member
        pstReply = (redisReply*)redisCommand(m_pstRedisCtx, "zrange %s -1 -1", szKey);
    }

    if (pstReply == NULL)
    {
        printf("zrange command failed: I/O error\n");
        return -1;
    }

    if (pstReply->type == REDIS_REPLY_ERROR) 
    {
        printf("zrange failed: %s\n",pstReply->str);
        return -1;
    }

    printf("%d:%s\n", pstReply->type, pstReply->str);

    if (pstReply->type == REDIS_REPLY_ARRAY)
    {
        printf("size: %lu\n", pstReply->elements);
        for (size_t i = 0; i < pstReply->elements; i++)
        {
            printf("%lu:%d:%s\n", i, pstReply->element[i]->type, pstReply->element[i]->str);
        }
    }

    printf("zadd zrange successfully\n");
}

int32_t CRedisClient::GetMarketData(int32_t iTradingDay, int32_t iBeginTime, int32_t iEndTime, std::vector<tick>& refTickVec)
{
    if (m_atmState.load() != RedisClientState::REDIS_STATE_CONNECTED)
    {
        return -1;
    }
}


int32_t CRedisClient::__Connect()
{
    m_atmState.store(RedisClientState::REDIS_STATE_DISCONNECT);
    
    if (m_pstRedisCtx != nullptr)
    {
        if (redisReconnect(m_pstRedisCtx) != REDIS_OK)
        {
            printf("Failed to connect redis, host=%s, port=%hu\n", m_strHost.c_str(), m_usPort);
            return -1;
        }
    }
    else
    {
        m_pstRedisCtx = redisConnectWithTimeout(m_strHost.c_str(), m_usPort, m_stTimeOut);
        if (m_pstRedisCtx == nullptr)
        {
            printf("Failed to connect redis, host=%s, port=%hu\n", m_strHost.c_str(), m_usPort);
            return -1;
        }

        if (m_pstRedisCtx->err != 0)
        {
            printf("Failed to connect redis, host=%s, port=%hu, err=%d, errstr=%s\n",
                    m_strHost.c_str(), m_usPort, m_pstRedisCtx->err, m_pstRedisCtx->errstr);
            return -1;
        }
    }
    
    printf("Connected to redis %s:%hu successfully\n", m_strHost.c_str(), m_usPort);
    m_atmState.store(RedisClientState::REDIS_STATE_CONNECTED);

    return 0;
}

int32_t CRedisClient::__SendHeartBeat()
{
    redisReply* pstReply;
    {
        std::unique_lock<std::mutex> objUL(m_objMtx);
        pstReply = (redisReply*)redisCommand(m_pstRedisCtx, "PING");
    }
    
    if (pstReply == NULL)
    {
        printf("PING command failed: I/O error\n");
        return -1;
    }

    if (pstReply->type == REDIS_REPLY_ERROR) 
    {
        printf("PING failed: %s\n",pstReply->str);
        freeReplyObject(pstReply);
        return -1;
    }

    freeReplyObject(pstReply);

    return 0;
}

void CRedisClient::__ThreadConnectMgrFunc()
{
    printf("Connection manager thread is running\n");

    prctl(PR_SET_NAME, "conn_mgr");

    printf("__ThreadEventFunc Thread: %lx\n", pthread_self());
        
    while (m_bConnMgrRunning)
    {
        std::this_thread::sleep_for(std::chrono::milliseconds(1000));

        if (m_atmState.load() == RedisClientState::REDIS_STATE_CONNECTED)
        {
            // 心跳发送失败，重连
            if (__SendHeartBeat() == 0)
            {
                printf("Sent heartbeat successfully\n");
                continue;
            }
        }

        printf("Reconnecting to redis %s:%hu\n", m_strHost.c_str(), m_usPort);
        (void)__Connect();
    }

    printf("Connection manager thread will exit\n");
}
